package org.rency.mq.activemq.jms.support;

import org.rency.mq.activemq.constant.DestinationType;
import org.springframework.jms.JmsException;
import org.springframework.jms.connection.ConnectionFactoryUtils;
import org.springframework.jms.core.MessageCreator;
import org.springframework.jms.core.SessionCallback;
import org.springframework.jms.support.JmsUtils;
import org.springframework.util.Assert;

import javax.jms.*;

/**
 * ${todo}
 *
 * @author: Created by user_rcy@163.com on 2016/10/19.
 * @version: V1.0.0
 * @since JDK 1.8
 */
public class JmsTemplate extends org.springframework.jms.core.JmsTemplate{

    /**
     * 创建JMS连接Session
     * @param connection JMS连接
     * @param transacted 是否支持事务
     * @param acknowledgeMode 应答模式
     * @return JMS连接Session
     * @throws JMSException
     */
    protected Session createSession(Connection connection, boolean transacted,int acknowledgeMode) throws JMSException {
        return connection.createSession(transacted, acknowledgeMode);
    }

    public Object execute(SessionCallback action, boolean transacted,int acknowledgeMode, boolean startConnection) throws JmsException {
        Assert.notNull(action, "Callback object must not be null");
        Connection conToClose = null;
        Session sessionToClose = null;
        try {
            conToClose = createConnection();
            sessionToClose = createSession(conToClose);
            if (startConnection) {
                conToClose.start();
            }
            logger.debug("Executing callback on JMS Session: " + sessionToClose);
            return action.doInJms(sessionToClose);
        } catch (JMSException ex) {
            throw convertJmsAccessException(ex);
        } finally {
            JmsUtils.closeSession(sessionToClose);
            ConnectionFactoryUtils.releaseConnection(conToClose, getConnectionFactory(), startConnection);
        }
    }

    public void send(final String destinationName, final int destinationType, boolean transacted,
                     int acknowledgeMode, final int deliveryMode, final MessageCreator messageCreator) throws JmsException {
        execute(new SessionCallback() {

            public Object doInJms(Session session) throws JMSException {
                Destination destination = resolveDestinationName(session, destinationName, destinationType);
                doSend(session, destination, messageCreator, deliveryMode);
                return null;
            }
        }, transacted, acknowledgeMode, false);
    }

    protected void doSend(Session session, Destination destination,MessageCreator messageCreator, int deliveryMode) throws JMSException {
        Assert.notNull(messageCreator, "MessageCreator must not be null");
        MessageProducer producer = createProducer(session, destination);
        try {
            Message message = messageCreator.createMessage(session);
            logger.debug("Sending created message: " + message);
            producer.send(message, deliveryMode, getPriority(), getTimeToLive());

            // Check commit - avoid commit call within a JTA transaction.
            if (session.getTransacted() && isSessionLocallyTransacted(session)) {
                // Transacted session created by this template -> commit.
                JmsUtils.commitIfNecessary(session);
            }
        }
        finally {
            JmsUtils.closeMessageProducer(producer);
        }
    }

    public Message receiveSelected(final String destinationName, final int destinationType, boolean transacted,int acknowledgeMode, final String messageSelector) throws JmsException {
        return (Message)execute(new SessionCallback() {

            public Message doInJms(Session session) throws JMSException {
                Destination destination = resolveDestinationName(session, destinationName, destinationType);
                return doReceive(session, destination, messageSelector);
            }

        }, transacted, acknowledgeMode, false);
    }

    protected Destination resolveDestinationName(Session session, String destinationName, int destinationType) throws JMSException {
        boolean pubSubDomain = destinationType == DestinationType.TOPIC;
        return getDestinationResolver().resolveDestinationName(session, destinationName, pubSubDomain);
    }

}